package com.liaofan.rocketmq.transaction;

import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.client.producer.TransactionMQProducer;
import org.apache.rocketmq.common.message.Message;

import java.util.concurrent.*;

public class TransacationProducer {
    public static void main(String[] args) throws MQClientException {
        TransactionMQProducer producer = new TransactionMQProducer("tpg");
        producer.setNamesrvAddr("192.168.153.129:9876");

        /**
         * 定义个线程池
         * Creates a new ThreadPoolExecutor with the given initial parameters and default rejected execution handler.
         * 参数:
         * corePoolSize 指定核心线程的数量
         * maximumPoolSize – 最多线程数
         * keepAliveTime – 当线程池中线程数量大于核心线程数量时,多余空闲线程的存活时间
         * unit – 时间单位
         * workQueue – 临时存放的任务队列,其参数就是队列的长度
         * threadFactory – 线程工厂
         * 抛出:
         * IllegalArgumentException – if one of the following holds: corePoolSize < 0 keepAliveTime < 0 maximumPoolSize <= 0 maximumPoolSize < corePoolSize
         * NullPointerException – if workQueue or threadFactory is null
         */
        ExecutorService executorService = new ThreadPoolExecutor(2, 5, 100, TimeUnit.SECONDS, new ArrayBlockingQueue<Runnable>(2000), new ThreadFactory() {
            @Override
            public Thread newThread(Runnable r) {
                Thread thread = new Thread(r);
                thread.setName("client- transaction-msg-check- thread");
                return thread;
            }
        });

        //为生产者添加
        producer.setExecutorService(executorService);
        producer.setTransactionListener(new ICBCTransacationListener());
        producer.start();

        String[] tags = {"TAGA", "TAGB", "TAGC"};

        for (int i = 0; i < 3; i++) {
            byte[] body = ("Hi," + i).getBytes();
            Message msg = new Message("somTopic_Transacation", tags[i], body);
            //发送事务消息
            //第二个参数用于指定在执行本地事务时要使用的业务参数
            SendResult sendResult = producer.sendMessageInTransaction(msg, null);
            System.out.println("发送结果为: " + sendResult.getSendStatus());

        }
    }
}
